/*******************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Co.,Ltd.        */
/*                                                                 */
/*  NAME      = frcmd_server.c                                     */
/*  NOTE      =                                                    */
/*  DATE      = 2014/11/17 by guojq                                */
/*******************************************************************/
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <stdlib.h>
#include <errno.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <pthread.h>
#include <zmq.h>

#include <string>

#include "frcomm.h"
#include "frmem.h"
#include "frlog.h"
#include "frame.h"
#include "frxml_parse.h"
#include "client.pb.h"
#include "frfilter.h"
#include "frcmd_server.h"
#include "frcmd_server_l7.h"
#include "frproto_l7default.h"
#include "frproto_mng.h"
#include "frmodule_mng.h"
#include "conntrack.h"
#include "frame_params.h"
#include "frct_comm.h"
#include "frct_mng.h"
#include "frct_hash.h"
#include "frstat_module.h"
#include "frstat_ct.h"
#include "frstat_system.h"
#include "frstat_thread.h"
#include "frstat_topology.h"
#include "frstat_server.h"
#include "frreport_mng.h"
#include "frlua_config.h"
#include "frssl_config.h"
#include "frexport_pcap.h"
#include "frshm_stat.h"

#define COMMIT_CMD             "commit"
#define ROLLBACK_CMD           "rollback"

#define FILTER_CMD             "filter"

#define FILTER_CMD_ADD         "add"
#define FILTER_CMD_DEL         "del"
#define FILTER_CMD_CLEAR       "clear"

#define PRINT_CMD_SYSTEM       "system"
#define PRINT_CMD_CT           "ct"
#define PRINT_CMD_TOPOLOGY     "topology"
#define PRINT_CMD_TASK         "task"
#define PRINT_CMD_MODULE       "module"
#define PRINT_CMD_SERVER       "server"
#define PRINT_CMD_FILTER       "filter"
#define PRINT_CMD_L7           "l7"
#define PRINT_CMD_JAVA         "java"
#define PRINT_CMD_MEMORY       "memory"
#define PRINT_CMD_VERSION      "version"
#define PRINT_CMD_SHM          "share_memory"

FR_STATIC pthread_rwlock_t g_cmd_rwlock;
FR_STATIC void *context = NULL;

union int_arr4 {
    uint32_t ui_len;
    char arr_data[4];
};

/*******************************************************************************
*  FUNC     :  执行print的命令
*              print system/ct/topology/thread/module/server/filter/l7_default/java/memory [L7_protocol]
*  ARGS     :  p_command(命令行)
*  RTN      :  FR_SUCC (success)
*           :  FR_FAIL (failed)
*  NOTE     :
*******************************************************************************/
FR_STATIC int __do_print(Command *p_command, char **pp_out, void **pp_accept_sock)
{
    PrintCommand st_printcmd = p_command->print_cmd();
    char *p_out = NULL;
    int i_ret = FR_FAIL;
    char *p_ret_tmp1 = NULL;
    char *p_ret_tmp2 = NULL;
    void *accept_sock = *pp_accept_sock;

    if (!st_printcmd.has_cmd())
        return FR_FAIL;
    
    //system
    if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_SYSTEM, strlen(PRINT_CMD_SYSTEM)) == 0)
    {
        frstat_system_output(pp_out);
        i_ret = FR_SUCC;
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_CT, strlen(PRINT_CMD_CT)) == 0)
    {
        frdbg("run frct_stat_all...\n");
        //frct_hash_output(&p_out);
        if(st_printcmd.args_size() == 0)
            frstat_ct_all(&p_out, accept_sock);
        else
        {
            if (strcasecmp(st_printcmd.args(0).c_str(), "all") == 0)
                frstat_ct_l7(pp_out, -1, pp_accept_sock);
            else if (strcasecmp(st_printcmd.args(0).c_str(), "unknown") == 0)
                frstat_ct_l7(pp_out, -2, pp_accept_sock);
            else
                frstat_ct_l7(pp_out, atoi(st_printcmd.args(0).c_str()), pp_accept_sock);
        }
        i_ret = FR_SUCC;
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_TOPOLOGY, strlen(PRINT_CMD_TOPOLOGY)) == 0)
    {
        if (frstat_topology_report(accept_sock) != FR_SUCC)
        {
            p_out = (char *)frmalloc(128);

            snprintf(p_out, 128, "print topology failed\n");
            *pp_out = p_out;
        }
        i_ret = FR_SUCC;
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_TASK, strlen(PRINT_CMD_TASK)) == 0)
    {
        frstat_thread_output(pp_out);
        i_ret = FR_SUCC;
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_MODULE, strlen(PRINT_CMD_MODULE)) == 0)
    {
        if(st_printcmd.args_size() > 0)
        {
            if(strncasecmp(st_printcmd.args(0).c_str(),"config", strlen("config"))==0)
            {
                frmodule_mng_output(&p_ret_tmp1);
                frproto_l7default_output(&p_ret_tmp2);
                if(p_ret_tmp1 != NULL && p_ret_tmp2 != NULL)
                {
                    //p_ret_tmp1 p_ret_tmp2  '\n'  '\0'
                    p_out = (char *) frmalloc(strlen(p_ret_tmp1) + strlen(p_ret_tmp2) + 2);
                    if(!p_out)
                    {
                        frfree(p_ret_tmp1);
                        frfree(p_ret_tmp2);
                        return FR_FAIL;
                    }

                    snprintf(p_out,strlen(p_ret_tmp1) + strlen(p_ret_tmp2) + 2,"%s\n%s",p_ret_tmp1,p_ret_tmp2);
                    frfree(p_ret_tmp1);
                    frfree(p_ret_tmp2);
                    *pp_out = p_out;
                    i_ret = FR_SUCC;
                }
                else if(p_ret_tmp1 != NULL)
                {
                    *pp_out = p_ret_tmp1;
                    i_ret = FR_SUCC;
                }
                else if(p_ret_tmp2 != NULL)
                {
                    *pp_out = p_ret_tmp2;
                    i_ret = FR_SUCC;
                }
                else
                    i_ret = FR_FAIL;
            }
            else if(strncasecmp(st_printcmd.args(0).c_str(),"stat", strlen("stat"))==0)
            {
                frstat_module_output(pp_out);
                i_ret = FR_SUCC;
            }
        }
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_SERVER, strlen(PRINT_CMD_SERVER)) == 0)
    {
        frstat_server_output(&p_out, accept_sock);
        i_ret = FR_SUCC;
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_FILTER, strlen(PRINT_CMD_FILTER)) == 0)
    {
        i_ret = apm_filter_show(pp_out);
        //frdbg("apm_filter_show \n[%s], [%d]\n", p_out, i_ret);
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_L7, strlen(PRINT_CMD_L7)) == 0)
    {
        frproto_define_output(pp_out);
        i_ret = FR_SUCC;
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_JAVA, strlen(PRINT_CMD_JAVA)) == 0)
    {
        if(st_apm_params.arr_zmq_connect[0] != '\0')
        {
            p_out = (char *) frmalloc(strlen(st_apm_params.arr_zmq_connect) + 1);
            if(!p_out)
                return FR_FAIL;

            snprintf(p_out,strlen(st_apm_params.arr_zmq_connect) + 1,"%s",st_apm_params.arr_zmq_connect);
            *pp_out = p_out;
            i_ret = FR_SUCC;
        }
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_MEMORY, strlen(PRINT_CMD_MEMORY)) == 0)
    {
        fr_slab_stat(pp_out);
        i_ret = FR_SUCC;
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_VERSION, strlen(PRINT_CMD_VERSION)) == 0)
    {
        const char *p_version = frame_version();
        if (p_version)
        {
            p_out = frstrdup(p_version);
            *pp_out = p_out;
            i_ret = FR_SUCC;
        }
    }
    else if (strncasecmp(st_printcmd.cmd().c_str(), PRINT_CMD_SHM, strlen(PRINT_CMD_SHM)) == 0)
    {
        frshm_stat_info(pp_out);
        i_ret = FR_SUCC;
    }
    else
        return FR_FAIL;

    return i_ret;
}

/*******************************************************************************
*  FUNC     :  执行commit的命令
*  ARGS     :  p_command(命令行)
*  RTN      :  FR_SUCC (success)
*           :  FR_FAIL (failed)
*  NOTE     :
*******************************************************************************/
FR_STATIC int __do_commit(Command *p_command, char **pp_result)
{
    CommitCommand st_commit_cmd = p_command->commit_cmd();
    int i_action = st_commit_cmd.action();
    int i_cmd = st_commit_cmd.cmd();

    switch (i_cmd)
    {
        case CommitCommand_Type_FILTER:
            if (i_action == CommitCommand_Action_Commit)
            {
                frdbg("apm_filter_commit start\n");
                apm_filter_commit(pp_result);
                return FR_SUCC;
            }
            else if (i_action == CommitCommand_Action_Rollback)
            {
                frdbg("apm_filter_rollback start\n");
                apm_filter_rollback();
                return FR_SUCC;
            }
            return FR_FAIL;
        case CommitCommand_Type_L7:
            if (i_action == CommitCommand_Action_Commit)
            {
                frdbg("apm_filter_commit start\n");
                return do_l7_commit(pp_result);
            }
            else if (i_action == CommitCommand_Action_Rollback)
            {
                frdbg("apm_filter_rollback start\n");
                do_l7_rollback();
                return FR_SUCC;
            }
            return FR_FAIL;
        case CommitCommand_Type_L7_DEFAULT:
            if (i_action == CommitCommand_Action_Commit)
            {
                frdbg("do_l7_default_commit start\n");
                return do_l7_default_commit(pp_result);
            }
            else if (i_action == CommitCommand_Action_Rollback)
            {
                frdbg("do_l7_default_rollback start\n");
                do_l7_default_rollback();
                return FR_SUCC;
            }
            return FR_FAIL;
        case CommitCommand_Type_MODULE:
            if (i_action == CommitCommand_Action_Commit)
            {
                frdbg("do_module_commit start\n");
                return do_module_commit(pp_result);
            }
            else if (i_action == CommitCommand_Action_Rollback)
            {
                frdbg("do_module_rollback start\n");
                do_module_rollback();
                return FR_SUCC;
            }
            return FR_FAIL;
        case CommitCommand_Type_LUA:
            if (i_action == CommitCommand_Action_Commit)
            {
                frdbg("do_lua_commit start\n");
                return do_lua_commit(pp_result);
            }
            else if (i_action == CommitCommand_Action_Rollback)
            {
                frdbg("do_lua_rollback start\n");
                do_lua_rollback();
                return FR_SUCC;
            }
            return FR_FAIL;
        case CommitCommand_Type_SSL:
            if (i_action == CommitCommand_Action_Commit)
            {
                frdbg("do_ssl_commit start\n");
                return do_ssl_commit(pp_result);
            }
            else if (i_action == CommitCommand_Action_Rollback)
            {
                frdbg("do_ssl_rollback start\n");
                do_ssl_rollback();
                return FR_SUCC;
            }
            return FR_FAIL;
    }
    
    return FR_SUCC;
}

/*******************************************************************************
*  FUNC     :  执行filter的命令
*  ARGS     :  p_command(命令行)
*           :  pp_result     错误原因
*  RTN      :  FR_SUCC (success)
*           :  FR_FAIL (failed)
*  NOTE     :
*******************************************************************************/
FR_STATIC int __do_filter(Command *p_command, char **pp_result)
{
    Filter st_filter;
    FilterCommand st_filter_cmd = p_command->filter_cmd();
    int i_ret = 0;

    if (!st_filter_cmd.has_action())
        goto err;

    if (strncasecmp(st_filter_cmd.action().c_str(), FILTER_CMD_CLEAR, strlen(FILTER_CMD_CLEAR)) == 0)
    {
        FRLOG_INFO("apm_filter_clear start\n");
        i_ret = apm_filter_clear();
        goto end;
    }

    if (p_command->filter_cmd().filters_size() <= 0)
        goto err;

    st_filter = p_command->filter_cmd().filters(0);

    if (strncasecmp(st_filter_cmd.action().c_str(), FILTER_CMD_ADD, strlen(FILTER_CMD_ADD)) == 0)
    {
        FRLOG_INFO("apm_prepare_add start, mac=[%s], proto=[%d], sip=[%s], sport=[%s], dip=[%s], dport=[%s]\n"
            , st_filter.mac().c_str(), st_filter.ip_proto()
            , st_filter.sip().c_str(), st_filter.sport().c_str(), st_filter.ip().c_str(), st_filter.port().c_str());
        i_ret = apm_prepare_add(1, st_filter.accept(), st_filter.mac().c_str(), st_filter.ip_proto()
                    , st_filter.sip().c_str(), st_filter.ip().c_str(), st_filter.sport().c_str(), st_filter.port().c_str());
    }
    else
        goto err;

end:
    if (i_ret != FR_SUCC)
        *pp_result = frstrdup(ERR_MALLOC_FAILED);
    
    return i_ret;
err:
    *pp_result = frstrdup(FRCMD_SERVER_NOT_SUPPORT);
    return FR_FAIL;
}

/*******************************************************************************
*  FUNC     :  退出程序处理
*  ARGS     :  
*  RTN      :  
*  NOTE     :
*******************************************************************************/
void __do_exit(void)
{
    //退出处理
    g_i_flag_run = 0;
    FRLOG_INFO("exit command");
}

FR_STATIC int __do_report(Command *p_command, char **pp_result)
{
    char arr_report_detail[128];

    if (p_command->report_cmd().has_console())
    {
        snprintf(arr_report_detail, sizeof(arr_report_detail), "console=%d",p_command->report_cmd().console());
    }
    else if (p_command->report_cmd().has_testtools())
    {
        snprintf(arr_report_detail, sizeof(arr_report_detail), "testtools=%d",p_command->report_cmd().testtools());
    }
    else if (p_command->report_cmd().has_writefile())
    {
        snprintf(arr_report_detail, sizeof(arr_report_detail), "writefile=%d",p_command->report_cmd().writefile());
    }
    else if (p_command->report_cmd().has_base64())
    {
        snprintf(arr_report_detail, sizeof(arr_report_detail), "base64=%d",p_command->report_cmd().base64());
    }
    else
    {
        *pp_result = frstrdup(FRCMD_SERVER_NOT_SUPPORT);
        return FR_FAIL;
    }

    frreport_debug(arr_report_detail);

    return FR_SUCC;
}

FR_STATIC int __do_lua(Command *p_command, char **pp_result)
{
    int i_ret = 0;

    if (p_command == NULL || !p_command->has_lua_cmd() || !p_command->lua_cmd().has_action())
        return FR_FAIL;

    FRLOG_INFO_DETAIL("start do lua %s", p_command->lua_cmd().action().c_str());
    if (strcasecmp(p_command->lua_cmd().action().c_str(), "add") == 0)
        i_ret = frlua_config_add(p_command->lua_cmd().ip(), p_command->lua_cmd().port().c_str(), p_command->lua_cmd().lua().c_str(), pp_result);
    else if (strcasecmp(p_command->lua_cmd().action().c_str(), "del") == 0)
        i_ret = frlua_config_del(p_command->lua_cmd().ip(), p_command->lua_cmd().port().c_str(), pp_result);
    else if (strcasecmp(p_command->lua_cmd().action().c_str(), "clear") == 0)
        i_ret = frlua_config_clear(pp_result);
    else
    {
        *pp_result = frstrdup(FRCMD_SERVER_NOT_SUPPORT);
        return FR_FAIL;
    }

    return i_ret;
}

FR_STATIC int __do_ssl(Command *p_command, char **pp_result)
{
    int i_ret = 0;

    if (p_command == NULL || !p_command->has_ssl_cmd())
        return FR_FAIL;

    if (strcasecmp(p_command->ssl_cmd().action().c_str(), "add") == 0)
        i_ret = frssl_config_add(p_command->ssl_cmd().ip(), p_command->ssl_cmd().port()
                , p_command->ssl_cmd().cert().c_str(), p_command->ssl_cmd().cert_content().c_str()
                , p_command->ssl_cmd().passwd().c_str(), p_command->ssl_cmd().domain().c_str()
                , pp_result);
    else if (strcasecmp(p_command->ssl_cmd().action().c_str(), "del") == 0)
        i_ret = frssl_config_del(p_command->ssl_cmd().ip(), p_command->ssl_cmd().port()
                , p_command->ssl_cmd().domain().c_str(), pp_result);
    else
    {
        *pp_result = frstrdup(FRCMD_SERVER_NOT_SUPPORT);
        return FR_FAIL;
    }

    return i_ret;
}

FR_STATIC int __do_clear_all(char **pp_result)
{
    int i_ret = 0;

    //清空L7信息
    i_ret = do_l7_clear_all(pp_result);
    if (i_ret != FR_SUCC)
        return i_ret;
    //清空模块
    i_ret = do_module_clear_all(pp_result);
    if (i_ret != FR_SUCC)
        return i_ret;
    //清空过滤规则
    i_ret = apm_filter_clear_all(pp_result);
    if (i_ret != FR_SUCC)
        return i_ret;
    //清空lua信息
    i_ret = frlua_clear_all(pp_result);
    if (i_ret != FR_SUCC)
        return i_ret;
    //清空ssl信息
    i_ret = do_ssl_clear_all(pp_result);
    if (i_ret != FR_SUCC)
        return i_ret;

    return FR_SUCC;
}

/*******************************************************************************
*  FUNC     :  根据收到的命令做相应操作
*  ARGS     :  p_command(命令行)
*           :  pp_result(结果集)
*  RTN      :  FR_SUCC (success)
*           :  FR_FAIL (failed)
*  NOTE     :
*******************************************************************************/
FR_STATIC int __do_command(void *p_cmd
                         , int i_cmd_len
                         , char **pp_out
                         , int *p_i_out
                         , void **accept_sock)
{
    int i_ret = FR_SUCC;
    Command st_command;
    CommandResult *p_result;
    std::string str_out;
    char *p_out = NULL;
    char *p_reason = NULL;

    p_result = new CommandResult();

    st_command.ParseFromArray(p_cmd,i_cmd_len);

    switch (st_command.cmd_type())
    {
        case Command_CommandType_Type_ModuleCommand:
            i_ret = do_module_cmd(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_L7Command:
            i_ret = do_l7_cmd(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_L7Default:
            i_ret = do_l7default_cmd(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_FilterCommand:
            i_ret = __do_filter(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_CommitCommand:
            i_ret = __do_commit(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_PrintCommand:
            i_ret = __do_print(&st_command, &p_out, accept_sock);
            if (p_out)
            {
                *pp_out = p_out;
                *p_i_out = strlen(p_out);
            }
            return i_ret;
        case Command_CommandType_Type_ExitCommand:
            __do_exit();
            break;
        case Command_CommandType_Type_ReportCommand:
            i_ret = __do_report(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_LuaCommand:
            i_ret = __do_lua(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_SslCommand:
            i_ret = __do_ssl(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_JavaCommand:
            //todo
            break;
        case Command_CommandType_Type_ExportCommand:
            i_ret = do_export_cmd(&st_command, &p_reason);
            break;
        case Command_CommandType_Type_ClearCommand:
            i_ret = __do_clear_all(&p_reason);
            break;
        default:
            goto err;
    }
    
    if (i_ret == FR_SUCC)
        p_result->set_status_code(CommandResult_ResultType_Type_OK);
    else
    {
        p_result->set_status_code(CommandResult_ResultType_Type_ERR);
        if (p_reason)
            p_result->set_reason(p_reason);
    }
    //把命令运行的结果做成protobuf数据
    p_result->SerializeToString(&str_out);
    if (str_out.length())
    {
        p_out = (char *)frmalloc(str_out.length());
        memcpy(p_out, str_out.c_str(), str_out.length());
        *pp_out = p_out;
        *p_i_out = str_out.length();
    }
    delete p_result;
    
    if (p_reason)
        frfree(p_reason);

    return FR_SUCC;
err:
    //命令不支持
    p_result->set_status_code(CommandResult_ResultType_Type_ERR);
    p_result->set_reason(FRCMD_SERVER_NOT_SUPPORT);

    p_result->SerializeToString(&str_out);
    if (str_out.length())
    {
        p_out = (char *)frmalloc(str_out.length());
        memcpy(p_out, str_out.c_str(), str_out.length());
        *pp_out = p_out;
        *p_i_out = str_out.length();
    }
    delete p_result;
    
    return FR_SUCC;
}

FR_STATIC void *__zmq_server_init(char *p_path)
{
    int i_timeout = 5000;
    void *socket = NULL;
    int i_ret = 0;

    int i_flag = 1;
    int i_keepalive_idle = 10;
    int i_keepalive_cnt = 1;
    int i_keepalive_intvl = 2;

    socket = zmq_socket(context, ZMQ_REP);
    if (socket == NULL)
    {
        FRLOG_ERROR_DETAIL("zmq_socket failed, errno=[%s]", strerror(errno));
        return NULL;
    }
    i_ret = zmq_bind(socket, p_path);
    if (i_ret != 0)
    {
        FRLOG_ERROR_DETAIL("zmq_bind failed, errno=[%s]", strerror(errno));
        zmq_close(socket);
        return NULL;
    }
    if(zmq_setsockopt(socket, ZMQ_LINGER, &i_timeout, sizeof(i_timeout)) < 0)
    {
        FRLOG_ERROR_DETAIL("zmq_setsockopt failed, errno=[%s]", strerror(errno));
        zmq_close(socket);
        return NULL;
    }
    if(zmq_setsockopt(socket, ZMQ_RCVTIMEO, &i_timeout, sizeof(i_timeout)) < 0)
    {
        FRLOG_ERROR_DETAIL("zmq_setsockopt failed, errno=[%s]", strerror(errno));
        zmq_close(socket);
        return NULL;
    }
    if(zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE, &i_flag, sizeof(i_flag)) < 0)
    {
        FRLOG_ERROR_DETAIL("zmq_setsockopt ZMQ_TCP_KEEPALIVE failed, errno=[%s]", strerror(errno));
        zmq_close(socket);
        return NULL;
    }
    if(zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_IDLE, &i_keepalive_idle, sizeof(i_keepalive_idle)) < 0)
    {
        FRLOG_ERROR_DETAIL("zmq_setsockopt ZMQ_TCP_KEEPALIVE_IDLE failed, errno=[%s]", strerror(errno));
        zmq_close(socket);
        return NULL;
    }
    if(zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_CNT, &i_keepalive_cnt, sizeof(i_keepalive_cnt)) < 0)
    {
        FRLOG_ERROR_DETAIL("zmq_setsockopt ZMQ_TCP_KEEPALIVE_CNT failed, errno=[%s]", strerror(errno));
        zmq_close(socket);
        return NULL;
    }
    if(zmq_setsockopt(socket, ZMQ_TCP_KEEPALIVE_INTVL, &i_keepalive_intvl, sizeof(i_keepalive_intvl)) < 0)
    {
        FRLOG_ERROR_DETAIL("zmq_setsockopt ZMQ_TCP_KEEPALIVE_INTVL failed, errno=[%s]", strerror(errno));
        zmq_close(socket);
        return NULL;
    }

    return socket;
}

FR_STATIC void __zmq_destroy(void *socket)
{
    zmq_close(socket);
    zmq_term(context);
    return;
}

/*******************************************************************************
*  FUNC     :  发送数据
*  ARGS     :  sockfd(socket句柄)
*           :  p_data(要发送的数据)
*           :  i_data_len(数据长度)
*           :  i_flag_more(是否需要继续发送)
*  RTN      :  FR_SUCC (success)
*           :  FR_FAIL (failed)
*  NOTE     :
*******************************************************************************/
int frzmq_send_msg(void **pp_socket, const char *p_data, int i_data_len, int i_flag_more)
{
    int i_ret = 0;
    int i_flag = 0;
    int i_max = 1023, i_send = 0;
    void *socket = *pp_socket;

    while (i_max < i_data_len)
    {
        i_flag = ZMQ_SNDMORE;
        zmq_send (socket, p_data + i_send, i_max, i_flag);
        i_send += i_max;
        i_data_len -= i_max;
    }
    i_flag = i_flag_more == 1?ZMQ_SNDMORE:0;

    i_ret = zmq_send (socket, p_data + i_send, i_data_len, i_flag);
    if (i_ret < 0)
    {
        if (errno == 11)
        {
            __zmq_destroy(socket);
            zmq_term(context);
            context = zmq_init(1);
            socket = __zmq_server_init(st_apm_params.arr_cmd_srv);
            *pp_socket = socket;
        }
        return FR_FAIL;
    }
    
    return FR_SUCC;
}

FR_STATIC int __zmq_recv_msg(void *socket, zmq_msg_t *p_zmq_message)
{
    int i_ret = FR_SUCC;

    i_ret = zmq_recvmsg(socket, p_zmq_message, 0);
    if (i_ret < 0 && errno != EAGAIN)
    {
        FRLOG_ERROR_DETAIL("cmd_server zmq recv message error,[%s], %d\n", strerror(errno), errno);
        return FR_FAIL;
    }
    
    return FR_SUCC;
}

/*******************************************************************************
*  FUNC     :  主程序
*  ARGS     :  argc(入力参数个数)
*           :  argv(入力参数)
*  RTN      :  FR_SUCC (success)
*           :  FR_FAIL (failed)
*  NOTE     :
*******************************************************************************/
int frcmd_server_start(void)
{
    int i_ret = 0;
    void *socket = NULL;

    if (frcmd_server_l7_init() != FR_SUCC)
    {
        FRLOG_ERROR_DETAIL("frcmd_server_l7_init failed\n");
        return FR_FAIL;
    }

    i_ret = pthread_rwlock_init(&g_cmd_rwlock, NULL);
    if (i_ret != 0)
    {
        FRLOG_ERROR_DETAIL("pthread_rwlock_init failed\n");
        frcmd_server_l7_destory();
        return FR_FAIL;
    }
    //初始化zmq
    context = zmq_init(1);
    if (context == NULL)
    {
        FRLOG_ERROR_DETAIL("zmq_ctx_new failed\n");
        pthread_rwlock_destroy(&g_cmd_rwlock);
        frcmd_server_l7_destory();
        return FR_FAIL;
    }
    
    //初始化服务端socket
    socket = __zmq_server_init(st_apm_params.arr_cmd_srv);
    if (socket == NULL)
    {
        zmq_term(context);
        pthread_rwlock_destroy(&g_cmd_rwlock);
        frcmd_server_l7_destory();
        return FR_FAIL;
    }
    while (g_i_flag_run)
    {
        zmq_msg_t st_zmq_message;
        char *p_send_msg = NULL;
        int i_send_len = 0;

        i_ret = zmq_msg_init(&st_zmq_message);
        if (i_ret != 0)
            goto err;

        if (__zmq_recv_msg(socket, &st_zmq_message) == FR_FAIL)
            goto err;
        
        //timeout继续做
        if (zmq_msg_size(&st_zmq_message) == 0)
            continue;
        
        i_ret = __do_command(zmq_msg_data(&st_zmq_message), zmq_msg_size(&st_zmq_message), &p_send_msg, &i_send_len, &socket);
        if (i_ret == FR_SUCC && p_send_msg)
            frzmq_send_msg(&socket, p_send_msg, i_send_len, 0);
        else
        {
            //str.append("CMD ERR\n");
            p_send_msg = frstrdup("CMD ERR\n");
            frzmq_send_msg(&socket, p_send_msg, i_send_len, 0);
            frfree(p_send_msg);
            zmq_msg_close(&st_zmq_message);
            continue;
        }
        zmq_msg_close(&st_zmq_message);
        frfree(p_send_msg);
    }

    __zmq_destroy(socket);
    zmq_term(context);
    pthread_rwlock_destroy(&g_cmd_rwlock);
    frcmd_server_l7_destory();

    return FR_SUCC;
err:
    __zmq_destroy(socket);
    zmq_term(context);
    pthread_rwlock_destroy(&g_cmd_rwlock);
    frcmd_server_l7_destory();
    return FR_FAIL;
}
